package com.example.demo.hello;

import com.alibaba.fastjson2.JSONObject;
import lombok.extern.slf4j.Slf4j;
import org.springframework.batch.core.*;
import org.springframework.batch.core.configuration.annotation.EnableBatchProcessing;
import org.springframework.batch.core.configuration.annotation.JobBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepBuilderFactory;
import org.springframework.batch.core.configuration.annotation.StepScope;
import org.springframework.batch.core.job.DefaultJobParametersValidator;
import org.springframework.batch.core.launch.support.RunIdIncrementer;
import org.springframework.batch.core.scope.context.ChunkContext;
import org.springframework.batch.core.step.tasklet.Tasklet;
import org.springframework.batch.item.*;
import org.springframework.batch.repeat.RepeatStatus;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.annotation.Bean;
import org.springframework.context.annotation.Configuration;

import java.util.List;
import java.util.Map;
import java.util.concurrent.atomic.AtomicInteger;

@Slf4j
@Configuration
@EnableBatchProcessing
public class HelloJob {
    @Autowired
    public JobBuilderFactory jobBuilderFactory;

    @Autowired
    public StepBuilderFactory stepBuilderFactory;

    private AtomicInteger count = new AtomicInteger(0);

    @Bean
    public Job importUserJob(Step step1,Step step2,DefaultJobParametersValidator validator) {
        return jobBuilderFactory.get("job8").validator(validator)
                .incrementer(new RunIdIncrementer()) // 增量修改参数，防止因为任务名称和参数一致而报错
                .listener(new JobExecutionListener(){
                    @Override
                    public void beforeJob(JobExecution jobExecution) {
                        log.info("作业状态：{}",jobExecution.getStatus());
                    }

                    @Override
                    public void afterJob(JobExecution jobExecution) {
                        log.info("作业状态：{}",jobExecution.getStatus());
                    }
                })
                .start(step2)
//                .next(step2)
                .build();
    }

    @Bean
    public Step step2(Tasklet tasklet) {
        return stepBuilderFactory.get("step2")
                .listener(new StepExecutionListener() {
                    @Override
                    public void beforeStep(StepExecution stepExecution) {
                        log.error("step开始");
                    }

                    @Override
                    public ExitStatus afterStep(StepExecution stepExecution) {
                        log.error("step结束");
                        return ExitStatus.COMPLETED;
                    }
                })
//                .tasklet(tasklet)
                .<String,String>chunk(3)
                .reader(itemReader())
                .processor(itemProcessor())
                .writer(itemWriter())
                .build();
    }

    @Bean
    public Step step1() {
        return stepBuilderFactory.get("step1").tasklet(new Tasklet() {
            @Override
            public RepeatStatus execute(StepContribution contribution, ChunkContext chunkContext) throws Exception {
                // 步骤上下文
                StepExecution stepExecution = chunkContext.getStepContext().getStepExecution();
                ExecutionContext executionContext = stepExecution.getExecutionContext();
                executionContext.put("step1","zhangsan");

                // job上下文
                ExecutionContext jobExecutionContext = chunkContext.getStepContext().getStepExecution().getJobExecution().getExecutionContext();
                jobExecutionContext.put("step1","lisi");
                return RepeatStatus.FINISHED;
            }
        }).build();
    }

    @Bean
    @StepScope
    public Tasklet tasklet(@Value("#{jobParameters['name']}") String name){
        return (stepContribution, chunkContext) -> {
            ExecutionContext executionContext = chunkContext.getStepContext().getStepExecution().getExecutionContext();
            log.error("步骤上下文步骤一的数据: {}",executionContext.get("step1"));

            Map<String, Object> jobExecutionContext = chunkContext.getStepContext().getJobExecutionContext();
            log.error("Job下文步骤一的数据: {}",jobExecutionContext.get("step1"));
            log.info("作业执行中状态：{}",stepContribution.getStepExecution().getJobExecution().getStatus());
            System.out.println("作业参数:  " + chunkContext.getStepContext().getJobParameters());
            return RepeatStatus.FINISHED;
        };
    }

    /**
     *  参数校验器
      * @return
     */
    @Bean
    public DefaultJobParametersValidator defaultJobParametersValidator(){
        DefaultJobParametersValidator validator = new DefaultJobParametersValidator();
        validator.setRequiredKeys(new String[]{"name","run.id"});
        validator.setOptionalKeys(new String[]{"age"});
        return validator;
    }

    @Bean
    public ItemReader<String> itemReader(){
        return new ItemReader<String>() {
            @Override
            public String read() throws Exception, UnexpectedInputException, ParseException, NonTransientResourceException {
                if(count.get()>10){
                    return null;
                }
                log.info("read....{}","a");
                count.incrementAndGet();
                return "a";
            }
        };
    }

    @Bean
    public ItemProcessor<String,String> itemProcessor(){
        return new ItemProcessor<String,String>() {
            @Override
            public String process(String o) throws Exception {
                log.info("process: {}",o);
                return o;
            }
        };
    }

    @Bean
    public ItemWriter<String> itemWriter(){
        return new ItemWriter<String>() {
            @Override
            public void write(List<? extends String> list) throws Exception {
                log.info("write: {}", JSONObject.toJSONString(list));
            }
        };
    }
}
